package com.hecore.lagou.hdfs;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;


/**
 * @Author: hecore
 * @Date: 2020/10/12 12:56
 * @Description:
 */

public class HdfsClientSample {


    public static String baseHost="hdfs://192.168.56.101:9000";

    /**
     * IO 操作必须关闭流. 1.1 文件夹创建
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    @Test
    public void testMkdirs() throws URISyntaxException, IOException, InterruptedException {
        // get file sys
        Configuration config=new Configuration();
        // execute by cluster
        // FileSystem fs = FileSystem.get(new URI("hdfs://192.168.56.101:9000"),config);
        FileSystem fs = FileSystem.get(new URI(baseHost),config,"root");
        fs.mkdirs(new Path("/test"));
        // fs.mkdirs(new Path("/test"));
        fs.close();
    }

    /**
     * listFile&Dir
     */
    @Test
    public void testListStatus() throws IOException, InterruptedException,
            URISyntaxException{

        // 1 获取⽂件配置信息
        FileSystem fs = getBaseFileSystem();
        // 2 判断是⽂件还是⽂件夹
        FileStatus[] listStatus = fs.listStatus(new Path("/"));

        for (FileStatus fileStatus : listStatus) {

            // 如果是⽂件
            if (fileStatus.isFile()) {
                System.out.println("f:"+fileStatus.getPath().getName());
            }else {
                System.out.println("d:"+fileStatus.getPath().getName());
            }
            // 以上我们使⽤的API操作都是HDFS系统框架封装好的。我们⾃⼰也可以采⽤IO流的⽅式实现⽂件的上传和下载。
        }
        // 3 关闭资源
        fs.close();
    }

    /**
     * listFile
     * recursive: 递推。
     */

    @Test
    public void listFiles() throws IOException {
        FileSystem fs=getBaseFileSystem();
        RemoteIterator<LocatedFileStatus> remoteIterator=fs.listFiles(new Path("/"),true);
        // 迭代器的while
        while (remoteIterator.hasNext()){
            // 下一个是
            LocatedFileStatus fileStatus=remoteIterator.next();
            final String file_name=fileStatus.getPath().getName();
            System.out.println(file_name);
            final long len=fileStatus.getLen();
            FsPermission permission =fileStatus.getPermission();
            // 存储块信息
            BlockLocation locations[] =fileStatus.getBlockLocations();
            final long accessTime=fileStatus.getAccessTime();
            for (BlockLocation location : locations) {
                //
                String[] hosts=location.getHosts();
                System.out.println(StringUtils.join(hosts, ","));
            }

        }
        fs.close();
    }

    /**
     * 删除文件
     *      删除不存在的也不会报错.
     */
    @Test
    public void testDelete(){
        try {

            deleteHdfs("/sample.txt",true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public FileSystem getBaseFileSystem(){
        Configuration configuration = new Configuration();
        FileSystem fs = null;
        try {
            fs = FileSystem.get(new URI(baseHost),
                    configuration, "root");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
        return fs;
    }


    /**
     * 删除文件/文件夹
     * @param file
     * @param isClear 内部文件是否清楚
     */
    private void deleteHdfs(String file,boolean isClear) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(baseHost),
                configuration, "root");
        fs.delete(new Path(file),isClear);
        fs.close();
    }


    /**
     * 上传文库
     */
    @Test
    public void testUpload()  throws Exception{
        copyFromLocalFile("F:/book/sample.txt","/sample.txt");
    }

    /**
     * 下载文件
     */
    @Test
    public void testDown() throws Exception {
        copyToLocalFile(false,"/sample.txt","/F:book/新版",false);
    }


    /**
     * 下载操作
     * @param delSrc 指是否将原⽂件删除
     * @param srcPath 指要下载的⽂件路径
     * @param dstPath 指将⽂件下载到的路径
     * @param isCheck 是否开启⽂件校验
     */
    public void copyToLocalFile(boolean delSrc, String srcPath,String dstPath,boolean isCheck) throws Exception {
        // basic
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI(baseHost),
                configuration, "root");
        // down
        fs.copyToLocalFile(delSrc, new Path(srcPath), new Path(dstPath), isCheck);
        // 3 关闭资源
        fs.close();
    }


    /**
     * 复制本地环境到hdfs
     * @param localFile
     * @param hdfsFile
     * @throws Exception
     */
    public void copyFromLocalFile(String localFile,String hdfsFile) throws URISyntaxException, IOException, InterruptedException {
        Configuration config=new Configuration();
        // 指定副本数量
        config.set("dfs.replation","2");
        FileSystem fs = FileSystem.get(new URI(baseHost),config,"root");
        fs.copyFromLocalFile(new Path(localFile),new Path(hdfsFile));
        fs.close();
    }

    @Test
    public void testException() throws IOException {
        Configuration config=new Configuration();
        config.set("fs.defaultFS",baseHost);
        FileSystem fs=FileSystem.get(config);
        fs.mkdirs(new Path("/test"));
        fs.close();
    }

}
